<?php

namespace Library\Rabbitmq;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

class Msg
{
    //连接rabbitMQ
    protected $connection ;
    //渠道
    protected $channel ;

    // 交换机类型
    protected $exchange = 'direct_msg'; 

    //队列名默认
    protected $queue = 'queue_default';

    public function __construct()
    {
        $this->connect();
        $this->queue_set();
    }

    public function connect()
    {
        $conf = [
            'host'=>'rabbitMq',
            'port'=>'5672',
            'user'=>'mq_hbw',
            'pwd'=>'mq_hbw',
            'vhost'=>'/',
        ];
        $this->connection = new AMQPStreamConnection( 
            $conf['host'], $conf['port'], $conf['user'], $conf['pwd'], $conf['vhost']
        );
        $this->channel = $this->connection->channel();
        return true;
    }

    /**
     * 交换机 队列初始化
     */
    public function queue_set()
    {
        // 声明一个队列
        // queue 队列名
        // passive 检测队列是否存在  true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // exclusive 私有队列 不允许其它用户访问  设置true 将会变成私有
        // auto_delete  当所有消费客户端连接断开后，是否自动删除队列
       // $this->channel->queue_declare($this->queue, false, false, false, false);

        // exchange 交换机名称
        // type 交换器类型 exchange types有四种：direct, topic, headers and fanout
        // passive 检测交换机是否存在   true 只检测不创建 false 创建
        // durable 是否持久化队列 true 为持久化
        // auto_delete  当所有绑定队列都不在使用时，是否自动删除交换器 true：删除false：不删除
        //$this->channel->exchange_declare($this->exchange, 'direct', false, false, false);

        // 绑定队列和交换机 $channel->queue_bind($queue_name, $exchange_name, $binding_key);
        //$this->channel->queue_bind($this->queue, $this->exchange);

    }

    public function publish($content, $exchange='', $routeKey='', $head = [])
    {
         
         // 写入队列的消息
         if( is_array($content) ){
            $messageBody = json_encode($content,JSON_UNESCAPED_UNICODE) ;
         }else{
             $messageBody = $content;
         }
         
        // 消息内容
        // delivery_mode  投递模式  delivery mode 设置为 2标记持久化
        // $message = new AMQPMessage($messageBody, array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        $head          = array_merge(array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT), $head);
        $message = new AMQPMessage($messageBody,$head);
         // $message 消息内容
         // $exchange  交换器名称
         // routing_key   路由键 (routing key)  主题交换机会用到
         $this->channel->basic_publish($message, $exchange,$routeKey);
         return true;
    }

    
    // public function consumer($queue, $consumerTag, $call_back)
    // {
    //     if (!isset($call_back) || empty($call_back)) {
    //         return false;
    //     }
 
    //     if (!$this->connect()) {
    //         return false;
    //     }
    //我们可以使用带有prefetch_count = 1设置的basic_qos方法 。这告诉 RabbitMQ 一次不要给一个 worker 多条消息。或者，换句话说，在工作人员处理并确认之前的消息之前，不要向工作人员发送新消息。相反，它将把它分派给下一个不忙的工人
    //     $this->channel->basic_qos(null, 10, null);

/*** 
    queue	队列名	 
    consumer_tag	消费者标签	 
    no_local	AMQP的标准，RabbitMQ没有实现	 
    no_ack	收到消息后是否不需要回复，即认为被消费	false
    exclusive	排他消费，即这个队列只能由一个消费者消费	false
    nowait	不返回执行结果	false
    callback	回调函数	null
    ticket	 	null
    arguments	参数	null
***/
 
    //     $this->channel->basic_consume($queue, $consumerTag, false, false, false, false, $call_back);
    //     while (count($this->channel->callbacks)) {
    //         $this->channel->wait();
    //     }
    // }


    public function channel_close(){
        // 关闭信道
        $this->channel->close();
    }

    public function connection_close(){
        //关闭 amqp 连接
        $this->connection->close();
    }

    public function __destruct()
    {
        $this->channel_close();
        $this->connection_close();
    }
}